package com.pine.dcts.task;

import com.pine.dcts.common.TaskStatusType;
import com.pine.dcts.config.DctsPropertis;
import com.pine.dcts.service.TaskRunnerCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author xiaoyuan
 * @create 2021/4/2 15:33
 **/
@Slf4j
public class DefaultTaskPool implements TaskPool {


    private ThreadPoolTaskExecutor executor;

    private Map<String, Future> jobPool;

    private TaskStatCollector taskStatCollector;

    private int size = 200;

    private ThreadPoolTaskExecutor futurePool;

    private DctsPropertis dctsPropertis;

    public DefaultTaskPool() {
        executor = new ThreadPoolTaskExecutor();
        //1: 核心线程数目
        executor.setCorePoolSize(2);
        //2: 指定最大线程数,只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(10);
        //3: 队列中最大的数目
        executor.setQueueCapacity(200);
        //4: 线程名称前缀
        executor.setThreadNamePrefix("job-");
        //5:当pool已经达到max size的时候，如何处理新任务
        // CallerRunsPolicy: 会在execute 方法的调用线程中运行被拒绝的任务,如果执行程序已关闭，则会丢弃该任务
        // AbortPolicy: 抛出java.util.concurrent.RejectedExecutionException异常
        // DiscardOldestPolicy: 抛弃旧的任务
        // DiscardPolicy: 抛弃当前的任务
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //6: 线程空闲后的最大存活时间(默认值 60),当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
        executor.setKeepAliveSeconds(60);
        //7:线程空闲时间,当线程空闲时间达到keepAliveSeconds(秒)时,线程会退出,直到线程数量等于corePoolSize,如果allowCoreThreadTimeout=true,则会直到线程数量等于0
        executor.setAllowCoreThreadTimeOut(false);
        executor.initialize();

        futurePool = new ThreadPoolTaskExecutor();
        //1: 核心线程数目
        futurePool.setCorePoolSize(2);
        //2: 指定最大线程数,只有在缓冲队列满了之后才会申请超过核心线程数的线程
        futurePool.setMaxPoolSize(10);
        //3: 队列中最大的数目
        futurePool.setQueueCapacity(200);
        //4: 线程名称前缀
        futurePool.setThreadNamePrefix("job-");
        //5:当pool已经达到max size的时候，如何处理新任务
        // CallerRunsPolicy: 会在execute 方法的调用线程中运行被拒绝的任务,如果执行程序已关闭，则会丢弃该任务
        // AbortPolicy: 抛出java.util.concurrent.RejectedExecutionException异常
        // DiscardOldestPolicy: 抛弃旧的任务
        // DiscardPolicy: 抛弃当前的任务
        futurePool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //6: 线程空闲后的最大存活时间(默认值 60),当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
        futurePool.setKeepAliveSeconds(60);
        //7:线程空闲时间,当线程空闲时间达到keepAliveSeconds(秒)时,线程会退出,直到线程数量等于corePoolSize,如果allowCoreThreadTimeout=true,则会直到线程数量等于0
        futurePool.setAllowCoreThreadTimeOut(false);
        futurePool.initialize();
        jobPool = new ConcurrentHashMap<>(size);
        taskStatCollector = new DefaultTaskStatCollector();
    }


    @Override
    public synchronized void submit(TaskRunner taskRunner) {
        TaskExecutorStat taskExecutorStat = null;

        taskExecutorStat = taskStatCollector.get(taskRunner.getTask().getTaskId());
        if (taskExecutorStat != null) {
            callback(taskRunner.getCallback(), TaskStatusType.NOT_START, "任务已存在");
            return;
        }
        if (taskStatCollector.taskSize() > dctsPropertis.getTask().getMaxNum()) {
            callback(taskRunner.getCallback(), TaskStatusType.NOT_START, "任务池已达到最大量");
            return;
        }

        Future future = executor.submit(taskRunner);
        callback(taskRunner.getCallback(), TaskStatusType.RUNNING, "任务运行中");
        taskStatCollector.pushTask(new TaskExecutorStat(future, taskRunner.getTask(), taskRunner.getCallback()));
        if (log.isDebugEnabled()) {
            log.debug("任务已提交成功！信息：{}", taskRunner.getTask().toString());
        }
        futurePool.submit(new TaskExecutorCallback(future, taskRunner.getCallback(), taskStatCollector));
    }

    @Override
    public void stop(String taskId) {
        Future future = jobPool.get(taskId);
        if (future != null) {
            future.cancel(true);
            taskStatCollector.remove(taskId);
        }
    }

    @Override
    public boolean isRunning(String taskId) {
        return taskStatCollector.get(taskId) != null;
    }

    @Override
    public void stopAll() {
        taskStatCollector.getAll().forEach(t -> {
            t.getFuture().cancel(true);
            callback(t.getCallback(), TaskStatusType.ERROR, "任务强行中断");
        });
        this.taskStatCollector.clear();
    }

    @Override
    public void shutdown() {
        stopAll();
        this.executor.shutdown();
        this.futurePool.shutdown();
    }

    private void callback(TaskRunnerCallback callback, TaskStatusType statusType, String message) {
        callback.setTaskStatus(statusType);
        callback.setMessage(message);
        callback.callback();
    }
}
